package com.semidata.trp.decision.job;

import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.rtd.core.http.server.HttpSession;
import com.semidata.rtd.core.log.Logger;
import com.semidata.rtd.engine.executor.TaskExecutor;
import com.semidata.rtd.engine.job.AbstractJob;
import com.semidata.rtd.engine.job.JobCompleteHandler;
import com.semidata.rtd.engine.job.Task;
import com.semidata.trp.decision.bid.DynamicBidTask;
import com.semidata.trp.decision.config.Config;
import com.semidata.trp.decision.filter.TaskFilter;
import com.semidata.trp.decision.frequency.TaskFrequency;
import com.semidata.trp.decision.request.Request;
import com.semidata.trp.decision.request.TaskRequest;
import com.semidata.trp.decision.response.Response;
import com.semidata.trp.decision.response.TaskResponse;
import com.semidata.trp.decision.select.TaskSelect;

public class DecisionJob extends AbstractJob {

	private static final long MAX_COUNT = Config.getConfig().getJobMaxCount();
	private static final ReentrantLock lock = new ReentrantLock();
	private static long count = 0;
	private static boolean active = true;
	private TaskRequest taskRequest ;
	private TaskFrequency taskFrequency;
	private TaskFilter taskFilter;
	private TaskSelect taskSelect;
	private TaskResponse taskResponse;
	private List<DynamicBidTask> tasks;
	private AtomicReference<List<Integer>> failedIndexs;
	private AtomicInteger dynamicTaskCount;
	private DecisionJobContext context;
//	private static long lastPrint = 0;

	public static DecisionJob newDecisionJob(TaskExecutor executor,
	        DecisionJobContext context) {

//		long now = System.currentTimeMillis();
//		if (now - lastPrint > 1000) {
//			System.out.println("job count = " + count);
//			lastPrint = now;
//		}

		boolean allowed = false;

		lock.lock();
		try {
			if (MAX_COUNT > 0 && count >= MAX_COUNT) {
				active = false;
			}

			if (active)
				count++;
			allowed = active;

		} finally {
			lock.unlock();
		}

		if (allowed)
			return new DecisionJob(executor, context);
		else
			return null;

	}

	private static void countDown() {

		lock.lock();
		try {
			count--;

			if (!active && count <= MAX_COUNT / 2) {
				active = true;
			}
		} finally {
			lock.unlock();
		}

	}

	private static class DoneHandler implements JobCompleteHandler {

		private DecisionJobContext context;

		public DoneHandler(DecisionJobContext context) {
			this.context = context;
		}

		@Override
		public void handle() {
			HttpSession session = context.getHttpSession();
			try {
				Request request = context.getRequest();
				StringBuffer sb = new StringBuffer();

				if (request == null) {
					Logger.log("ERROR", "can not parse request");
					session.getResponse().error(HttpResponseStatus.BAD_REQUEST);
					return;
				}
				sb.append(request.getUserId()).append('^')
				        .append(request.getPositionId()).append('^')
				        .append(request.getIp()).append('^')
				        .append(request.getArea() == null ? null : request.getArea().toString()).append('^')
				        .append(request.getTimestamp());
				Response response = context.getResponse();
				sb.append(context.getBusinessLog());
				String responseStr;
				if (request.getCallback() != null) {
					responseStr = request.getCallback() + "("
					        + response.toString() + ");";
					session.getResponse().setHeader(
					        HttpHeaders.Names.CONTENT_TYPE, "application/json");
				} else {
					responseStr = response.toString();
				}
				session.getResponse().end(responseStr);
				sb.append(' ').append(
				        System.currentTimeMillis() - request.getTimestamp());
				Logger.log("BUSINESS", sb.toString());
			} catch (Exception e) {
				Logger.log("ERROR", e);
				session.getResponse().error(HttpResponseStatus.BAD_REQUEST);
			} finally {
				countDown();
			}
		}

	}

	private static class ErrorHandler implements JobCompleteHandler {

		private DecisionJobContext context;

		public ErrorHandler(DecisionJobContext context) {
			this.context = context;
		}

		@Override
		public void handle() {
			HttpSession session = context.getHttpSession();
			try {

				Request request = context.getRequest();
				StringBuffer sb = new StringBuffer();
				if (request == null) {
					Logger.log("ERROR", "can not parse request");
					session.getResponse().error(HttpResponseStatus.BAD_REQUEST);
					return;
				}
				sb.append(request.getUserId()).append('^')
				        .append(request.getPositionId()).append('^')
				        .append(request.getIp()).append('^')
				        .append(request.getArea() == null ? null : request.getArea().toString()).append('^')
				        .append(request.getTimestamp()).append('^');
				String businesslog = context.getBusinessLog();
				if (businesslog != null)
					sb.append(businesslog);
				sb.append(' ').append(
				        System.currentTimeMillis() - request.getTimestamp());
				Logger.log("BUSINESS", sb.toString());
			} catch (Exception e) {
				Logger.log("ERROR", e);
			} finally {
				countDown();
				session.getResponse().error(HttpResponseStatus.BAD_REQUEST);
			}
		}

	}

	public DecisionJob(TaskExecutor executor, DecisionJobContext context) {

		super(executor, new DoneHandler(context), new ErrorHandler(context));
		this.context = context;
		taskRequest = new TaskRequest(this, 0);
		taskFrequency = new TaskFrequency(this,0);
		taskFilter = new TaskFilter(this, 0);
		taskSelect = new TaskSelect(this, 0);
		taskResponse = new TaskResponse(this, 0);
		tasks = new ArrayList<>();
		List<Integer> index = new ArrayList<>();
		failedIndexs = new AtomicReference<>();
		failedIndexs.set(index);
		dynamicTaskCount = new AtomicInteger(0);
	}

	private void logException(Task task, Exception e) {

		String str = "";

		if (e.getClass() == InterruptedException.class) {
			str = task.getClass().getName() + " time out";
		} else {
			str = task.getClass().getName() + " throw exception: "
			        + e.toString() + "\n";
			StringWriter content = new StringWriter();
			PrintWriter writer = new PrintWriter(content);
			e.printStackTrace(writer);
			str += content.toString();
		}

		Logger.log("ERROR", str);

	}

	// private static class SummaryTask extends SyncTask {
	//
	// public SummaryTask(Job job) {
	// super(job);
	// }
	//
	// @Override
	// public void run() throws Exception {
	// TaskFilter taskFilter = new TaskFilter(getJob());
	//
	// taskFilter.run();
	//
	// TaskSelect taskSelect = new TaskSelect(getJob());
	//
	// taskSelect.run();
	//
	// TaskResponse taskResponse = new TaskResponse(getJob());
	//
	// taskResponse.run();
	//
	// }
	//
	// }
	
	@Override
	protected void schedule(Task task) {
		TaskExecutor executor = getExecutor();
		Logger.log("DEBUG", "run");
		if (task == null) {
			Logger.log("DEBUG", "start a task request");
			executor.submit(taskRequest);
		} else if (task.getClass() == TaskRequest.class) {
			Exception cause = task.getException();
			if (cause != null) {
				logException(task, cause);
				error();
			} else {
				Logger.log("DEBUG", "start a task frequency");
				executor.submit(taskFrequency);
			}

		} else if (task.getClass() == TaskFrequency.class) {
			Logger.log("DEBUG", "start a task filter");
			Exception cause = task.getException();
			if (cause != null)
				logException(task, cause);
			executor.submit(taskFilter);
		} else if (task.getClass() == TaskFilter.class) {
			Exception cause = task.getException();
			if (cause != null) {
				logException(task, cause);
				error();
			} else {
				tasks = taskFilter.getTasks();
				if (tasks.size() == 0) {
					Logger.log("DEBUG", "start a task select");
					executor.submit(taskSelect);
				} else {
					for (DynamicBidTask dTask : tasks)
						executor.submit(dTask);
				}
			}
		} else if (task.getClass() == DynamicBidTask.class) {
			Exception cause = task.getException();
			if (cause != null) {
				List<Integer> indexs = failedIndexs.get();
				indexs.add(((DynamicBidTask)task).getIndex());
			}
			if (dynamicTaskCount.incrementAndGet() == tasks.size()) {
				taskSelect.setFailedIndex(failedIndexs.get());
				taskSelect.setFilter(taskFilter);
				executor.submit(taskSelect);
			}
		} else if (task.getClass() == TaskSelect.class) {
		
			Exception cause = task.getException();
			if (cause != null) {
				logException(task, cause);
				error();
			} else {
				Logger.log("DEBUG", "start a task response");
				executor.submit(taskResponse);
			}
		} else if (task.getClass() == TaskResponse.class) {
			Exception cause = task.getException();
			if (cause != null) {
				logException(task, cause);
				error();
			} else {
				Logger.log("DEBUG", "start a task done");
				done();
			}

		} else {
			error();
		}
	}

	public DecisionJobContext getJobContext() {
		return context;
	}

}
